package com.hexinfo.dmpro.data.transfer.common.service.impl;

import cn.hutool.core.util.StrUtil;
import cn.hutool.db.Db;
import cn.hutool.db.DbUtil;
import cn.hutool.db.ds.DSFactory;
import cn.hutool.db.handler.BeanListHandler;
import cn.hutool.db.sql.SqlExecutor;
import com.hexinfo.dmpro.data.transfer.common.model.CdhColumnsEntity;
import com.hexinfo.dmpro.data.transfer.common.model.CdhDbsEntity;
import com.hexinfo.dmpro.data.transfer.common.model.CdhFieldEntity;
import com.hexinfo.dmpro.data.transfer.common.model.CdhTblsEntity;
import com.hexinfo.dmpro.data.transfer.common.service.ICdhMetadataService;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.List;

/**
 * @Package: com.hexinfo.dmpro.data.transfer.common.service.impl
 * @ClassName CdhMetadataService
 * @Description: CDH元数据查询
 * @Author: wz.li
 * @Date 2023/11/24 14:40
 * @Version v2.0.2
 **/
@Service
public class CdhMetadataService implements ICdhMetadataService {

    @Value("${hive.metadata.version}")
    private String version;

    @Override
    public List<CdhDbsEntity> queryAllDatabase() {
        DataSource ds = DSFactory.get(version);
        Db.use(ds);
        Connection conn = null;
        try {
            conn = ds.getConnection();
            String querySql = "SELECT DB_ID, DB_LOCATION_URI, NAME FROM DBS WHERE NAME NOT LIKE 'uat_%' AND  NAME NOT IN ('default') ORDER BY NAME ASC";
            return (List<CdhDbsEntity>) SqlExecutor.query(conn, querySql, new BeanListHandler(CdhDbsEntity.class));
        } catch (SQLException e) {
            e.printStackTrace();
        } finally {
            DbUtil.close(conn);
        }
        return null;
    }

    @Override
    public List<CdhTblsEntity> queryTableListByDatabase(String database) {
        DataSource ds = DSFactory.get(version);
        Db.use(ds);
        Connection conn = null;
        try {
            conn = ds.getConnection();
            String tmplate = "SELECT t.TBL_ID, t.DB_ID, t.TBL_NAME FROM TBLS t LEFT JOIN DBS d  ON t.DB_ID = d.DB_ID WHERE d.NAME = '{}' and t.TBL_TYPE != 'VIRTUAL_VIEW' and not ( t.TBL_NAME like \n" +
                    "'%tinf\\_%' escape '\\' or t.TBL_NAME like '%\\_ext' escape '\\' or t.TBL_NAME like '%\\_test' escape \n" +
                    "'\\' or t.TBL_NAME like 'test\\_%' escape '\\' or t.TBL_NAME like '%\\_temp\\_%' escape '\\' or t.TBL_NAME \n" +
                    "like 'temp\\_%' escape '\\' or t.TBL_NAME like '%\\_temp' escape '\\' or t.TBL_NAME like '%\\_tmp\\_%' \n" +
                    "escape '\\' or t.TBL_NAME like '%\\_tmp' escape '\\' or t.TBL_NAME like 'tmp\\_%' escape '\\' or \n" +
                    "t.TBL_NAME like 'view\\_%' escape '\\' )  ORDER BY t.TBL_NAME ASC";
            String querySql = StrUtil.format(tmplate, database);
            return (List<CdhTblsEntity>) SqlExecutor.query(conn, querySql, new BeanListHandler(CdhTblsEntity.class));
        } catch (SQLException e) {
            e.printStackTrace();
        } finally {
            DbUtil.close(conn);
        }
        return null;
    }

    @Override
    public List<CdhFieldEntity> queryPartitionList(String database, String table) {
        DataSource ds = DSFactory.get(version);
        Db.use(ds);
        Connection conn = null;
        try {
            conn = ds.getConnection();
            String tmplate = "SELECT pk.PKEY_COMMENT AS FIELD_COMMENT, pk.PKEY_NAME AS FIELD_NAME, pk.PKEY_TYPE AS FIELD_TYPE, pk.INTEGER_IDX AS SEQ " +
                    "FROM PARTITION_KEYS pk " +
                    "LEFT JOIN TBLS t  ON pk.TBL_ID = t.TBL_ID " +
                    "LEFT JOIN DBS d ON t.DB_ID = d.DB_ID " +
                    "WHERE d.NAME = '{}' AND t.TBL_NAME = '{}' ORDER BY pk.INTEGER_IDX ASC";
            String querySql = StrUtil.format(tmplate, database, table);
            return (List<CdhFieldEntity>) SqlExecutor.query(conn, querySql, new BeanListHandler(CdhFieldEntity.class));
        } catch (SQLException e) {
            e.printStackTrace();
        } finally {
            DbUtil.close(conn);
        }
        return null;
    }

    @Override
    public List<CdhFieldEntity> queryColumnList(String database, String table) {
        DataSource ds = DSFactory.get(version);
        Db.use(ds);
        Connection conn = null;
        try {
            conn = ds.getConnection();
            String tmplate = "SELECT c.\"COMMENT\" AS FIELD_COMMENT, c.COLUMN_NAME AS FIELD_NAME, c.TYPE_NAME AS FIELD_TYPE, c.INTEGER_IDX AS SEQ " +
                    "FROM dbs d " +
                    "INNER JOIN tbls t " +
                    "ON d.db_id = t.db_id " +
                    "LEFT JOIN sds s " +
                    "ON t.sd_id = s.sd_id " +
                    "LEFT JOIN columns_v2 c " +
                    "ON s.cd_id = c.cd_id " +
                    "WHERE d.NAME = '{}' AND t.TBL_NAME = '{}' ORDER BY c.INTEGER_IDX ASC";
            String querySql = StrUtil.format(tmplate, database, table);
            return (List<CdhFieldEntity>) SqlExecutor.query(conn, querySql, new BeanListHandler(CdhFieldEntity.class));
        } catch (SQLException e) {
            e.printStackTrace();
        } finally {
            DbUtil.close(conn);
        }
        return null;
    }

    @Override
    public List<CdhColumnsEntity> queryTableByFieldName(String fieldName) {
        DataSource ds = DSFactory.get(version);
        Db.use(ds);
        Connection conn = null;
        try {
            conn = ds.getConnection();
            String tmplate = "SELECT D1.NAME AS DB_NAME, T1.TBL_NAME AS TABLE_NAME, C1.COLUMN_NAME AS FIELD_NAME, C1.TYPE_NAME AS FIELD_TYPE  \n" +
                    "FROM TBLS T1  \n" +
                    "LEFT JOIN DBS D1  \n" +
                    "  ON D1.DB_ID = T1.DB_ID  \n" +
                    "LEFT JOIN SDS S1\n" +
                    "  ON T1.SD_ID = S1.SD_ID\n" +
                    "LEFT JOIN COLUMNS_V2 C1\n" +
                    "  ON C1.CD_ID =S1.CD_ID  \n" +
                    "WHERE C1.COLUMN_NAME = '{}' \n" +
                    "  AND D1.NAME NOT LIKE 'uat_%'";
            String querySql = StrUtil.format(tmplate, fieldName);
            return (List<CdhColumnsEntity>) SqlExecutor.query(conn, querySql, new BeanListHandler(CdhColumnsEntity.class));
        } catch (SQLException e) {
            e.printStackTrace();
        } finally {
            DbUtil.close(conn);
        }
        return null;
    }
}
